package storm.online.bolt;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

import storm.online.common.LoginInfo;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

//这个用来判断是登陆还是登出
//解析json串
//查询当前账号，角色的信息，预估进入redis查询
public class LogBehaviorBolt extends BaseRichBolt{
	
	private OutputCollector collector;
	private Map<String, Integer> playerStatus = new HashMap<String, Integer>(); //这个用来储存玩家的状态，0表示离线，1表示在线
	
	
	private int onlineNumChange( String pt, Integer action ){
		int numChange = 0;
		
		if( playerStatus.containsKey(pt) ){
			int previosStatus = playerStatus.get(pt);
			if( previosStatus == 1 && action == 2 ){
				numChange = -1;
			}
			else if( previosStatus == 2 && action == 1 ){
				numChange = 1;
			}
		}
		else{
			if( action == 1 ){
				numChange = 1;
			}
		}

		playerStatus.put(pt, action);
		
		return numChange;
	}
	
	public void prepare(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		 this.collector = collector;		
	}

	public void execute(Tuple tuple) {
		
		String ptId = tuple.getStringByField("pt_id");
		Long areaId = tuple.getLongByField("area_id");
		String gameId = tuple.getStringByField("game_id");
		Long timestamp = tuple.getLongByField("timestamp");
		Integer action = tuple.getIntegerByField("action");
		
		int onlineNumChange = onlineNumChange( ptId, action );
		
		System.out.println("pt_id is " +  ptId + ", area_id is " + areaId + ", num_change is " + onlineNumChange);
		
		collector.emit(new Values(gameId, areaId, onlineNumChange));
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub
		declarer.declare( new Fields("game_id", "area_id", "num_change"));
	}
 
}
